package xxljob

import (
	"context"
	"fmt"
	"git.mycaigou.com/gfyx/common/cache"
	"git.mycaigou.com/gfyx/common/tenant_db"
	"git.mycaigou.com/gfyx/micro-gfyx-oppty-service/infrastructure/di"
	"git.mycaigou.com/gfyx/micro-gfyx-oppty-service/infrastructure/repository/remote/gfyx_user_service_remote"
	tenantPb "git.mycaigou.com/gfyx/micro-gfyx-oppty-service/proto/micro_gfyx_user_service_proto/micro-gfyx-user-service/tenant"
	"git.mycaigou.com/gfyx/micro-gfyx-oppty-service/service/base/profit"
	"git.mycaigou.com/gfyx/micro-gfyx-oppty-service/service/business_opportunity/subscription"
	"git.mycaigou.com/gfyx/micro-gfyx-oppty-service/service/clue/customer"
	"git.mycaigou.com/gfyx/micro-gfyx-oppty-service/service/job_center"
	"git.myscrm.cn/golang/stark/v4"
	"github.com/xxl-job/xxl-job-executor-go"
	"sync"
)

var executor xxl.Executor
var once sync.Once

const (
	serverAddr  = "xxljob.server_addr"
	accessToken = "xxljob.access_token"
	clientName  = "xxljob.client_name"
	clientPort  = "xxljob.client_port"

	radarPushMessageKey = "radar:message-push"
)

func InitExecutor() {
	defer func() {
		if err := recover(); err != nil {
			stark.Logger.Errorf(context.Background(), "xxl-job initExecutor error: %v", err)
		}
	}()
	go initExecutor()
}

func initExecutor() {
	once.Do(func() {
		serverAddr := stark.Configer.(stark.ApolloConfigerIface).GetStringValue(serverAddr, "")
		accessToken := stark.Configer.(stark.ApolloConfigerIface).GetStringValue(accessToken, "")
		port := stark.Configer.(stark.ApolloConfigerIface).GetStringValue(clientPort, "9999")
		name := stark.Configer.(stark.ApolloConfigerIface).GetStringValue(clientName, "micro-gfyx-oppty-service")
		executor = xxl.NewExecutor(
			xxl.ServerAddr(serverAddr),
			xxl.AccessToken(accessToken),
			xxl.ExecutorPort(port),
			xxl.RegistryKey(name),
		)
		executor.Init()

		// 中间件
		// executor.Use()

		// 日志
		// executor.LogHandler()

		// 注册任务
		executor.RegTask("continueAllJob", continueAllJob)
		executor.RegTask("sendRadarSubscriptionMessage", sendRadarSubscriptionMessage)
		executor.RegTask("calculateProfitAccountNum", calculateProfitAccountNum)
		executor.RegTask("updateOpenIdSubscribeStatus", updateOpenIdSubscribeStatus)
		executor.RegTask("updateCustomerData", updateCustomerData)
		err := executor.Run()
		if err != nil {
			stark.Logger.Errorf(context.Background(), "xxl-job Run error: %v", err)
		} else {
			stark.Logger.Infof(context.Background(), "xxl-job Run success")
		}
	})
}

func continueAllJob(ctx context.Context, param *xxl.RunReq) (msg string) {
	c := di.InjectInstance()
	_ = c.Invoke(func(
		jobService job_center.JobCenterServiceIface,
	) {
		stark.Logger.Infof(ctx, "continueAllJob param:%#v", param)
		go func() {
			defer func() {
				if err := recover(); err != nil {
					stark.Logger.Infof(ctx, "ReStartJob-err: %+v", err)
				}
			}()

			tenantList, err := gfyx_user_service_remote.NewGfyxUserServiceRepository().GetAllEfficientTenant(ctx, &tenantPb.GetAllEfficientTenantRequest{})
			//for test
			//tenantList = []*tenant_common.Tenant{{Code: "ycg"}}
			//for test end
			stark.Logger.Infof(ctx, "ReStartJob tenantList:%+v", tenantList)

			if err != nil {
				stark.Logger.Errorf(ctx, "ReStartJob GetAllEfficientTenant err:%s", err.Error())
				return
			}
			for _, tenantInfo := range tenantList.List {
				tenantCtx, _ := tenant_db.GetTenantBackgroundContext(ctx, tenantInfo.Code)
				if (&tenant_db.TenantDB{}).GetTenantDB(tenantCtx) == nil {
					stark.Logger.Errorf(tenantCtx, "ReStartJob tenantCode:%s,GetTenantDB failed", tenantInfo.Code)
					continue
				}

				var ids []int64
				ids, err = jobService.GetUnfinishedJobIds(tenantCtx)
				if err != nil {
					stark.Logger.Errorf(tenantCtx, "ReStartJob GetUnfinishedJobIds tenant_code:%s, err:%s", tenantInfo.Code, err.Error())
					continue
				}
				stark.Logger.Infof(tenantCtx, "ReStartJob tenant_code:%s, 未完成的任务id集合为：%+v", tenantInfo.Code, ids)
				for _, jobId := range ids {
					err = jobService.StartJob(tenantCtx, jobId)
					if err != nil {
						stark.Logger.Errorf(ctx, "ReStartJob StartJob tenant_code:%s, err:%s", tenantInfo.Code, err.Error())
						break
					}
				}
			}
		}()

	})
	return "success"
}

func sendRadarSubscriptionMessage(ctx context.Context, param *xxl.RunReq) (msg string) {
	c := di.InjectInstance()
	_ = c.Invoke(func(
		sub subscription.SubscriptionServiceIface,
	) {
		stark.Logger.Infof(ctx, "sendRadarSubscriptionMessage param:%#v", param)
		defer func() {
			if err := recover(); err != nil {
				stark.Logger.Infof(context.Background(), "cron-err: %+v", err)
			}
		}()

		cacheRedis, _ := cache.Redis()
		// 加锁，避免生产多个pod并发重复执行
		flag, err := cacheRedis.Incr(ctx, radarPushMessageKey)
		if err != nil {
			stark.Logger.Errorf(ctx, "sendRadarSubscriptionMessage redis Incr err: %s", err.Error())
			return
		}

		if flag > 1 {
			stark.Logger.Infof(ctx, "sendRadarSubscriptionMessage is running: %d", flag)
			return
		}

		// 设置过期时间
		_, err = cacheRedis.Expire(ctx, radarPushMessageKey, 60)
		if err != nil {
			stark.Logger.Errorf(ctx, "sendRadarSubscriptionMessage redis Expire err: %s", err.Error())
			return
		}
		err = sub.SendRadarSubscriptionMessage(ctx)
		if err != nil {
			stark.Logger.Errorf(ctx, "SendRadarSubscriptionMessage err: %s", err.Error())
		} else {
			stark.Logger.Infof(ctx, "SendRadarSubscriptionMessage success")
		}
	})
	return "success"
}

func calculateProfitAccountNum(ctx context.Context, param *xxl.RunReq) (msg string) {
	c := di.InjectInstance()
	_ = c.Invoke(func(
		profitService profit.ProfitServiceIface,
	) {
		stark.Logger.Infof(ctx, "calculateProfitAccountNum param:%#v", param)
		defer func() {
			if err := recover(); err != nil {
				stark.Logger.Infof(context.Background(), "calculateProfitAccountNum-err: %+v", err)
			}
		}()

		cacheRedis, _ := cache.Redis()
		cacheKey := "profit:calculate_account_num"
		// 加锁，避免生产多个pod并发重复执行
		flag, err := cacheRedis.Incr(ctx, cacheKey)
		if err != nil {
			return
		}

		if flag > 1 {
			stark.Logger.Infof(ctx, "calculateProfitAccountNum-is-running: %d", flag)
			return
		}

		// 设置过期时间
		_, _ = cacheRedis.Expire(ctx, cacheKey, 180)

		err = profitService.CalculateProfitAccountNum(ctx)
		if err != nil {
			stark.Logger.Errorf(ctx, "CalculateProfitAccountNum err: %s", err.Error())
		} else {
			stark.Logger.Infof(ctx, "CalculateProfitAccountNum success")
		}
	})
	return "success"
}

func updateOpenIdSubscribeStatus(ctx context.Context, param *xxl.RunReq) (msg string) {
	c := di.InjectInstance()
	_ = c.Invoke(func(
		subService subscription.SubscriptionServiceIface,
	) {
		stark.Logger.Infof(ctx, "updateOpenIdSubscribeStatus param:%#v", param)
		defer func() {
			if err := recover(); err != nil {
				stark.Logger.Infof(context.Background(), "updateOpenIdSubscribeStatus-err: %+v", err)
			}
		}()

		cacheRedis, _ := cache.Redis()
		cacheKey := "subscribe:ycg_open_id"
		// 加锁，避免生产多个pod并发重复执行
		flag, err := cacheRedis.Incr(ctx, cacheKey)
		if err != nil {
			return
		}

		if flag > 1 {
			stark.Logger.Infof(ctx, "updateOpenIdSubscribeStatus-is-running: %d", flag)
			return
		}

		// 设置过期时间
		_, _ = cacheRedis.Expire(ctx, cacheKey, 180)

		err = subService.UpdateOpenIdSubscribeStatus(ctx)
		if err != nil {
			fmt.Println("UpdateOpenIdSubscribeStatus err:", err.Error())
		} else {
			fmt.Println("UpdateOpenIdSubscribeStatus success")
		}
	})
	return "success"
}

func updateCustomerData(ctx context.Context, param *xxl.RunReq) (msg string) {
	c := di.InjectInstance()
	_ = c.Invoke(func(customerManageService customer.CustomerManageServiceIface) {
		stark.Logger.Infof(ctx, "updateCustomerData param:%#v", param)
		defer func() {
			if err := recover(); err != nil {
				stark.Logger.Infof(context.Background(), "updateCustomerData-err: %+v", err)
			}
		}()

		cacheRedis, _ := cache.Redis()
		cacheKey := "gfyx:cron_lock:customer_update"
		// 加锁，避免生产多个pod并发重复执行
		flag, err := cacheRedis.Incr(ctx, cacheKey)
		if err != nil {
			stark.Logger.Infof(ctx, "updateCustomerData-Incr failed: %v", err)
			return
		}

		if flag > 1 {
			stark.Logger.Infof(ctx, "updateCustomerData-is-running: %d", flag)
			return
		}

		// 设置过期时间
		_, _ = cacheRedis.Expire(ctx, cacheKey, 240)

		// 查找租户列表
		tenantList, err := gfyx_user_service_remote.NewGfyxUserServiceRepository().GetAllEfficientTenant(ctx, &tenantPb.GetAllEfficientTenantRequest{})
		for _, tenantInfo := range tenantList.List {
			tenantCtx, _ := tenant_db.GetTenantBackgroundContext(ctx, tenantInfo.Code)
			if (&tenant_db.TenantDB{}).GetTenantDB(tenantCtx) == nil {
				stark.Logger.Errorf(tenantCtx, "StartJob UpdateCustomerByCron tenantCode:%s,GetTenantDB failed", tenantInfo.Code)
				continue
			}
			result, err := customerManageService.UpdateCustomerByCron(tenantCtx)
			if err != nil {
				stark.Logger.Errorf(tenantCtx, "StartJob UpdateCustomerByCron tenant_code:%s, err:%s", tenantInfo.Code, err.Error())
				continue
			}
			msg += "租户：" + tenantInfo.Code + result + "\n"
		}
	})
	return msg
}
